package com.byron.media.server.ffmpeg;

import com.byron.media.server.utils.FrameInfo;
import lombok.AllArgsConstructor;
import lombok.Data;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 数据按照PTS定时输出
 */
public class PtsQueueSender implements Runnable {

    private int maxMills;            // 当已经收到最大毫秒数为max时候，暂停拉取

    private int minMills;            // 当已经收到最小毫秒数为min时候，就继续拉取

    private OverListener overListener;

    private FrameListener frameListener;

    private BlockingQueue<PtsFrameInfo> frameQueue;

    private boolean running;

    private long startTime;         // 开始时间（毫秒）

    public PtsQueueSender(int maxMills, int minMills, OverListener overListener, FrameListener frameListener){
        this.maxMills = maxMills;
        this.minMills = minMills;
        this.overListener = overListener;
        this.frameListener = frameListener;
        frameQueue = new LinkedBlockingQueue<>();
    }

    /**
     * 开始线程
     */
    public void start(){
        running = true;
        new Thread(this).start();
    }

    public void pushFrameInfo(FrameInfo frameInfo){

        // 如果推送的数据已经大于当前最大值，停止推送
        if(frameInfo.isVideo()){
            long mills = frameInfo.getPts() / 90 + startTime;
            frameQueue.offer(new PtsFrameInfo(mills, frameInfo));
            if(mills > (System.currentTimeMillis() + maxMills)){
                if(overListener != null){
                    overListener.onOverMaxMills();
                }
            }
        } else {
            // TODO 音频暂时不处理
        }
    }

    @Override
    public void run() {
        startTime = System.currentTimeMillis();
        while(running){
            PtsFrameInfo frameInfo = null;
            try {
                frameInfo = frameQueue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
                if(!running){
                    break;
                }
            }
            if(!running){
                break;
            }
            if(frameInfo != null){

                long now = System.currentTimeMillis();

                // 当包的时间小于最小时间，则继续播放
                if(frameInfo.getMillis() < now + minMills){
                    if(overListener != null){
                        overListener.onOverMinMills();
                    }
                }

                long delay = System.currentTimeMillis() - frameInfo.getMillis();
                if(delay > 0){
                    try {
                        Thread.sleep(delay);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                if(frameListener != null){
                    frameListener.onFrame(frameInfo.getFrameInfo());
                }
            }
        }

    }

    public interface OverListener {

        /**
         * 输入过度
         */
        void onOverMaxMills();

        /**
         * 输入过小
         */
        void onOverMinMills();
    }

    @Data
    @AllArgsConstructor
    public class PtsFrameInfo {

        private long millis;                // 播放时间

        private FrameInfo frameInfo;        // 源帧数据

    }
}
